package com.huawei.codearts.service;

import com.huawei.codearts.config.BackMQConfig;
import com.huawei.codearts.config.DelayMQConfig;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class MessageConsumer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 生产 > 发送消息
    public Object sendMessage(String message) {
        if (message == null) {
            message = "";
        }
        String uuid = UUID.randomUUID().toString();
        Map<String, Object> messageMap = new ConcurrentHashMap<>();
        messageMap.put("message", message);
        messageMap.put("uuid", uuid);
        rabbitTemplate.convertAndSend("requestQueue", messageMap);

        /**
         * String exchange, 交换机
         * String routingKey,  路由键，要与第四步绑定时设置的路由键有关系才能接收
         * Object object, 发送的对象，如果时对象那么这个对象必须是序列化的
         * @Nullable CorrelationData correlationData
         * 特别注意：（#）匹配0个或多个单词，（*） 匹配一个单词
         * 如果 routingKey 有test.# 则test.queue会写入3条数据
         */
        rabbitTemplate.convertAndSend("test-exchange",".queue",
                messageMap, new CorrelationData(uuid));

        rabbitTemplate.convertAndSend("test-exchange","test.queue.demo",
                messageMap, new CorrelationData(uuid));

        // 手动消息确认队列
        rabbitTemplate.convertAndSend("test-exchange","test.queue.confirm",
                messageMap, new CorrelationData(uuid));


        // ttl过期消息队列 方式一
        MessagePostProcessor processor = message1 -> {
            //设置message的信息，5秒后过期
            message1.getMessageProperties().setExpiration("5000");
            return message1;
        };
        //发送消息，此消息会独立过期
        rabbitTemplate.convertAndSend("test_exchange_ttl", "test_queue_ttl", "message ttl ...", processor);

        // ttl过期消息队列 方式二
        MessageProperties mps = new MessageProperties();
        mps.setExpiration("5000");
        /*
         * 这个参数见是用来做消息的唯一标识的
         * 发布消息时使用，存储在消息的headers中
         */
        CorrelationData cdata = new CorrelationData(UUID.randomUUID().toString());
        Message message2 = new Message("message ttl ...".getBytes(), mps);
        rabbitTemplate.convertAndSend("test_exchange_ttl", "test_queue_ttl", message2, cdata);


        // 测试优先级队列
        MessagePostProcessor messagePostProcessor = message3 -> {
            //设置消息优先级，值越大，越先消费
            message3.getMessageProperties().setPriority(5);
            return message3;
        };
        rabbitTemplate.convertAndSend("test_exchange_ttl", "", messageMap, messagePostProcessor);

        return messageMap;
    }

    // 生产 > 延迟消息 > demo 订单延迟关闭消息队列
    public Object delaySendMessage(String message) {
        if (message == null) {
            message = "";
        }
        String uuid = UUID.randomUUID().toString();
        Map<String, Object> messageMap = new ConcurrentHashMap<>();
        messageMap.put("message", message);
        messageMap.put("data", "订单信息：id=1001，time=2022年6月6日 13:00:00");
        messageMap.put("uuid", uuid);
        rabbitTemplate.convertAndSend(DelayMQConfig.ORDER_EXCHANGE_NAME,"order.info", messageMap, new CorrelationData(uuid));

        // 备份交换机
//        rabbitTemplate.setMandatory(true);
//        rabbitTemplate.setReturnsCallback(returned -> {
//            System.out.printf("消息被退回了....因为交换机 %s 上不存在routingKey：%s", returned.getExchange(), returned.getRoutingKey());
//        });
        rabbitTemplate.convertAndSend(BackMQConfig.CONFIRM_EXCHANGE_NAME, "unknown", "warning test ... ");

        return messageMap;
    }

    // 消费 > 监听消息
    @RabbitListener(queues = "requestQueue")
    public void handleMessage(String message) {
        // 处理消息的逻辑
        System.out.println("MessageConsumer..." + message + ", running:..." + Thread.currentThread().getName());

    }

    // 消费 > 监听消息
//    @RabbitHandler
    @RabbitListener(queues = {"test.queue"})
    public void handleMessageTest(Message message) {
        //拿到主体内容
        byte[] body = message.getBody();
        String bodyJson = new String(body);
        //拿到的消息头属性信息
        MessageProperties messageProperties = message.getMessageProperties();
        System.out.println("handleMessageTest..." + bodyJson + ", messageProperties:..." + messageProperties);

    }

    /**
     * queues：声明需要监听的队列
     * message ： 如果我们不知道发送的对象类型，可以统一用org.springframework.amqp.core.Message 来接收
     * OrderReturnReasonEntity:接收消息的对象，因为我们发的时这个对象，所以接收也可以用这个对象
     * Message message,OrderReturnReasonEntity 这两个参数写一个就可以
     * channel：com.rabbitmq.clientChannel 当前传输数据的通道
     * 场景：
     *      1）、订单服务启动多个，同一个信息，只能有一个客户端接收。
     *      2）、只有一个消息完全处理完，方法运行结束，我们才可以接收其他的消息。
     */
    @RabbitListener(queues = {"test-queue"})
//    @RabbitHandler
    public void revieveMessage(Message message, Channel channel) throws IOException {
        //拿到主体内容
        byte[] body = message.getBody();
        String bodyJson = new String(body);
        System.out.println("接受到的消息...内容" + bodyJson + " ===类型：" + message.getClass());
        //拿到的消息头属性信息
        MessageProperties messageProperties = message.getMessageProperties();
        System.out.println("接受到的消息...头属性" + messageProperties);
        // 在chanel内部是顺序自增的
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        System.out.println("maruis----deliveryTag-->" + deliveryTag);

        /**
         * deliveryTag 消息id
         * false   不批量
         */
        // 签收消息，非批量模式
        try {
//            if (deliveryTag % 2 == 0) {
                channel.basicAck(deliveryTag,false);
                System.out.println("maruis------>" + "签收成功");
//            } else {
//                // 参数： 消息id，是否重新投递给服务器
////                channel.basicReject(deliveryTag,false);
//                // 参数： 消息id，是否批量拒绝（如果为true，那么这条信息之前的信息全部被拒），是否重新投递给服务器（如果是true，就会重新放入队列，false丢弃）
//                channel.basicNack(deliveryTag,false,true);
//                System.out.println("maruis------>" + "退货");
//            }
        } catch (Exception e) {
            // 网络终端
            channel.basicNack(deliveryTag,false,true);
            System.out.println("maruis------>" + "退货");
            System.out.println("error------>" + e.getMessage());
        }

    }


    /**
     * 消费 > 监听队列
     * Consumer Ack机制：
     * 1、设置手动签收  acknowledge = manual
     * 2、让监听器实现ChannelAwareMessageListener接口或者使用 @RabbitListener注解
     * 3、如果消息成功处理，则调用 channel.basicAck()  签收
     * 4、如果消息处理失败，则调用 channel.basicNack() 拒绝签收，broker重新发送给consumer
     *
     * @param message
     * @param channel
     */
    @SneakyThrows
    @RabbitListener(queues = "test_queue_confirm")
    public void testManual(Message message, Channel channel) {
        Thread.sleep(1000);
        //获取消息传递的标记
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //接收消息
            System.out.println(new String(message.getBody()));
            //处理业务逻辑
            //例如，操作数据库，发送短信，等等...
            System.out.println("处理业务逻辑中...");
//            int i = 10 / 0;
            /*
             * 手动签收：
             * 第一个参数：表示收到的标签
             * 第二个参数：表示可以签收
             */
            channel.basicAck(deliveryTag, true);
            System.out.println("签收成功...");
        } catch (Exception e) {
            System.out.println(e.getMessage());
            //如果处理业务逻辑的时候发生了错误，则要拒绝签收
            //requeue = true 表示消息拒收，重新放回到队列中，并且一般放置在队列头部的位置
            // 如果为false，则不会放回到队列中，如果有死信队列则放入到死信队列，没有则丢弃
            channel.basicNack(deliveryTag, true, true);
            System.out.println("拒绝签收...");
        }
    }

    /**
     * 消费 > 延迟队列
     * 队列过期时间 delayMQConfig里已设置 也可以在生产者里为每个消息单独设置ttl过期时间
     * @param message
     * @param channel
     */
    @SneakyThrows
    @RabbitListener(queues = "order_queue_dlx")
    public void testDelayed(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收转换消息
            System.out.println("order_queue_dlx..." + new String(message.getBody()));
            //2. 处理业务逻辑
            System.out.println("order_queue_dlx...处理业务逻辑...");
            System.out.println("order_queue_dlx...根据订单id查询其状态...");
            System.out.println("order_queue_dlx...判断状态是否为支付成功");
            System.out.println("order_queue_dlx...取消订单,回滚库存....");
            //3. 手动签收
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            //e.printStackTrace();
            System.out.println("order_queue_dlx...出现异常,拒绝接受");
            //4.拒绝签收,不重回队列 requeue=false
            channel.basicNack(deliveryTag, true, false);
        }
    }

    /**
     * 备份交换机：
     * 交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout
     * 接收报警消息
     * @param message
     */
//    @SneakyThrows
//    @RabbitListener(queues = "warning.queue")
//    public void testWarning(Message message) {
//        String msg = new String(message.getBody());
//        System.out.printf("报警！！发现了不可路由消息：{%s}" + msg);
//    }

}
